package redis.gambing;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

import static redis.gambing.GambleFinalString.*;

/**
 * 计算总资产，赌博次数
 */
public class CalculatedGamblerInfo extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String gamblerId = input.getStringByField(GAMBLER_ID);
        Long money = input.getLongByField(HASH_GAMBLER_1_MONEY);
        Short times = input.getShortByField(HASH_GAMBLER_2_TIMES);
        Long asserts = input.getLongByField(HASH_GAMBLER_3_ASSERTS);

        if (input.getBooleanByField(GAMBLE_RESULT)) {
            asserts += money;
            money = 0L; // 只要赌资为0，不论是否赢了，都不会改变资产
        } else {
            asserts -= money;
            money += money;
            if (money > asserts) {
                money = 0L;
            }
        }
        times ++;
        collector.emit(new Values(gamblerId, money, times, asserts));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(GAMBLER_ID, HASH_GAMBLER_1_MONEY, HASH_GAMBLER_2_TIMES, HASH_GAMBLER_3_ASSERTS));
    }
}
